package com.demo.ordered;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.Charset;
import java.util.List;

public class MyMessageListenerConcurrently implements MessageListenerConcurrently {
    /*
    MessageListenerConcurrently 不能保证消费顺序，除非线程个数 <= 队列个数
     */
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(final List<MessageExt> msgs,
                                                    final ConsumeConcurrentlyContext context) {
        for (int i = 0; i < msgs.size(); i++) {
            System.out.printf(Thread.currentThread().getName() + " Receive New Messages: "
                    + new String(msgs.get(i).getBody(), Charset.forName("UTF-8")) + "%n");
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
